feat(source): add NATS source consumer parameters #17615

gbto · 2024-07-08T15:37:25Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR adds the 20 missing parameters supported in the async_nats crate of the consumer created in NATS sources. The DDL of a NATS source would now look like:

CREATE SOURCE (
...
) WITH (
    connector='nats',
    server_url='{{ env_var("SERVER") }}',
    subject='risingwave.test.source',
    stream='risingwave-test-source',
    scan.startup.mode='earliest',
    connect_mode='user_and_password',
    username='{{ env_var("USER") }}',
    password='{{ env_var("PASSWORD") }}',
    consumer.durable_name='risingwave-test-source',
    consumer.description='desc-test-source',
    consumer.ack_policy='all',
    consumer.ack_wait=10,
    consumer.max_deliver=10,
    consumer.filter_subjects='demo.subject.filter.*',
    consumer.filter_subjects='demo.subject.filter.1,demo.subject.filter.2',
    consumer.replay_policy='instant',
    consumer.rate_limit=100000000000,
    consumer.sample_frequency=100,
    consumer.max_waiting=10,
    consumer.max_ack_pending=10,
    -- consumer.idle_heartbeat=60, not available in async_nats crate
    consumer.max_batch=1000,
    consumer.max_bytes=1000000000,
    consumer.max_expires=3600,
    consumer.inactive_threshold=10000000,
    consumer.memory_storage='false',
    consumer.backoff='10,30,60',
    consumer.num_replicas=1
) FORMAT PLAIN ENCODE JSON

All parameters are optional. If left empty, it falls back to the async_nats default values. All are eventually parsed as strings in RisingWave and casted to the appropriate types before being passed to the build_consumer method.

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Release note

Now, all the parameters supported by the async_nats crate are supported in the RisingWave NATS source connector. In particular, users can now configure few important aspects of consumers such as the number of replicas that previously would default to 1, hereby causing Jetstream issues of synchronization across replicas. It also allows setting parameters such as max_waiting or max_ack_pending that can be critical for performance optimization of NATS clusters.

No backward incompatibility would be introduced by these changes.

BugenZhao · 2024-07-09T03:05:02Z

Hi. Can you help to rebase the commits first?

stdrc · 2024-07-10T02:54:27Z

src/connector/src/connector_common/common.rs

+            durable_name,
+            description,
+            ack_wait: ack_wait.unwrap_or_default(),
+            max_deliver: max_deliver.unwrap_or_default(),
+            filter_subject: filter_subject.unwrap_or_default(),
+            filter_subjects: filter_subjects.unwrap_or_default(),
+            rate_limit: rate_limit.unwrap_or_default(),
+            sample_frequency: sample_frequency.unwrap_or_default(),
+            max_waiting: max_waiting.unwrap_or_default(),
+            max_ack_pending: max_ack_pending.unwrap_or_default(),
+            // idle_heartbeat: idle_heart.unwrap_or_default(),
+            max_batch: max_batch.unwrap_or_default(),
+            max_bytes: max_bytes.unwrap_or_default(),
+            max_expires: max_expires.unwrap_or_default(),
+            inactive_threshold: inactive_threshold.unwrap_or_default(),
+            memory_storage: memory_storage.unwrap_or_default(),
+            backoff: backoff.unwrap_or_default(),
+            num_replicas: num_replicas.unwrap_or_default(),
            ..Default::default()


These .unwrap_or_default()s have subtly different semantics from ..Default::default(), although in current fact they behave the same because Config is just deriving Default but not manually implementing it.

I'm OK with this but we should be aware of the difference.

stdrc · 2024-07-10T03:04:23Z

src/connector/src/connector_common/common.rs

+        sample_frequency: Option<u8>,
+        max_waiting: Option<i64>,
+        max_ack_pending: Option<i64>,
+        _idle_heartbeat: Option<Duration>,


Is this deprecated? May just remove it.

Well the behavior of the missed idle heartbeat is a bit cryptic with the ‘async_nats’ crate and is a relatively important parameter user because it translates into error messages sent by the stream in the form of “missed idle heartbeat”, which can cause pipelines to fail. Not sure how to handle this one tbh

Removed from the configuration, since it doesn't exist in the pull consumer config: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/pull/struct.Config.html

tabVersion · 2024-07-10T03:39:46Z

Suggest following the practice at #11203
The introduce params are not essential to run NATS connector, so better to move them into a new struct.

xxchan · 2024-07-10T04:18:45Z

src/connector/src/source/nats/source/reader.rs

+                properties.ack_wait.clone().map(|s| {
+                    Duration::from_secs(s.parse::<u64>().expect("failed to parse ack_wait to u64"))
+                }),
+                properties.max_deliver.clone().map(|s| {
+                    s.parse::<i64>()
+                        .expect("failed to parse max_deliver to i64")
+                }),
+                properties.filter_subject.clone(),
+                properties
+                    .filter_subjects
+                    .clone()
+                    .map(|s| s.split(',').map(|s| s.to_string()).collect()),


These don't look good. We'd better use typed properties, and parse the config when creating the properties struct.
So that CREATE SOURCE will reject invalid config, instead of failing at runtime.

e.g., using #[serde_as(as = "Option<DisplayFromStr>")] or #[serde(deserialize_with = ...)] like other connectors

Ah indeed that seems much better, even without understanding the error related to config parsing that parsing seemed a bit sketchy. I’ll give it a try in the upcoming days @xxchan thank you

I've moved the consumer specific parameters into a separate struct, and added some typed properties based on @xxchan's suggestion

…ters

benjamin-awd · 2024-08-23T06:40:59Z

Just added a commit to address the comments from previous reviewers! Sorry for coming back late on this 🙏

stdrc

LGTM, can approve if others don't have further comments.

tabVersion

rest LGTM

src/connector/src/lib.rs

tabVersion · 2024-08-23T07:55:17Z

src/connector/src/source/nats/mod.rs

+    pub max_bytes: Option<i64>,
+
+    #[serde(rename = "consumer.max_expires")]
+    #[serde_as(as = "Option<DurationSeconds<String>>")]


src/connector/src/source/nats/mod.rs

tabVersion · 2024-08-23T08:01:29Z

src/connector/src/source/nats/mod.rs

+    #[serde(rename = "consumer.ack_policy")]
+    #[serde_as(as = "Option<DisplayFromStr>")]
+    pub ack_policy: Option<String>,


maybe we can try to deser it into Enum directly, serde can help us reject unrecognized str.

I tried to do this, but unfortunately couldn't find a clean way to do it -- the enums from the NATS crate do not implement FromStr, so it doesn’t seem possible to deserialize directly.

nevermind, I will try refactor it later. Thanks for your contribution.

tabVersion

LGTM, thanks

yufansong

LGTM

Co-authored-by: benjamin-awd <benjamindornel@gmail.com> (cherry picked from commit 220fded)

Co-authored-by: benjamin-awd <benjamindornel@gmail.com>

gbto requested a review from a team as a code owner July 8, 2024 15:37

gbto requested a review from MrCroxx July 8, 2024 15:37

graphite-app bot requested a review from a team July 8, 2024 15:39

stdrc changed the title ~~feat/add-nats-source-consumer-parameters~~ feat(source): add NATS source consumer parameters Jul 9, 2024

gbto force-pushed the feat/add-nats-consumer-parameters branch from e3ce905 to 6f3f8f9 Compare July 9, 2024 08:53

feat: Add 20 missing NATS consumer parameters

90a675b

gbto force-pushed the feat/add-nats-consumer-parameters branch from e044338 to 90a675b Compare July 9, 2024 08:58

stdrc requested review from stdrc, tabVersion and yufansong and removed request for a team and MrCroxx July 10, 2024 02:32

stdrc approved these changes Jul 10, 2024

View reviewed changes

xxchan reviewed Jul 10, 2024

View reviewed changes

gbto and others added 2 commits August 23, 2024 08:12

Merge branch 'risingwavelabs:main' into feat/add-nats-consumer-parame…

742c4db

…ters

refactor: create separate struct for NATS consumer properties

25494aa

fuyufjh requested review from stdrc and xxchan August 23, 2024 06:52

stdrc reviewed Aug 23, 2024

View reviewed changes

tabVersion reviewed Aug 23, 2024

View reviewed changes

benjamin-awd added 2 commits August 25, 2024 12:14

refactor: raise error if policy is invalid

d5f0858

refactor: rename time-related properties to *.sec

f54cc24

tabVersion approved these changes Aug 25, 2024

View reviewed changes

yufansong approved these changes Aug 25, 2024

View reviewed changes

fuyufjh added this pull request to the merge queue Aug 26, 2024

Merged via the queue into risingwavelabs:main with commit 220fded Aug 26, 2024
28 of 30 checks passed

benjamin-awd mentioned this pull request Aug 27, 2024

fix(source/nats): add missing default attribute for vectors #18261

Merged

9 tasks

zwang28 pushed a commit that referenced this pull request Aug 27, 2024

feat(source): add NATS source consumer parameters (#17615)

06c1b26

Co-authored-by: benjamin-awd <benjamindornel@gmail.com> (cherry picked from commit 220fded)

kwannoel pushed a commit that referenced this pull request Aug 27, 2024

feat(source): add NATS source consumer parameters (#17615)

80ff556

Co-authored-by: benjamin-awd <benjamindornel@gmail.com>

BugenZhao mentioned this pull request Dec 11, 2024

risingwave 2.1.0 risingwavelabs/homebrew-risingwave#46

Closed

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(source): add NATS source consumer parameters #17615

feat(source): add NATS source consumer parameters #17615

gbto commented Jul 8, 2024 •

edited by tabVersion

Loading

BugenZhao commented Jul 9, 2024

stdrc Jul 10, 2024

stdrc Jul 10, 2024

gbto Jul 10, 2024

benjamin-awd Aug 23, 2024

tabVersion commented Jul 10, 2024

xxchan Jul 10, 2024 •

edited

Loading

gbto Jul 10, 2024

benjamin-awd Aug 23, 2024

benjamin-awd commented Aug 23, 2024

stdrc left a comment

tabVersion left a comment

tabVersion Aug 23, 2024

tabVersion Aug 23, 2024

benjamin-awd Aug 25, 2024

tabVersion Aug 25, 2024

tabVersion left a comment

yufansong left a comment

feat(source): add NATS source consumer parameters #17615

feat(source): add NATS source consumer parameters #17615

Conversation

gbto commented Jul 8, 2024 • edited by tabVersion Loading

What's changed and what's your intention?

Checklist

Release note

BugenZhao commented Jul 9, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion commented Jul 10, 2024

xxchan Jul 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benjamin-awd commented Aug 23, 2024

stdrc left a comment

Choose a reason for hiding this comment

tabVersion left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion left a comment

Choose a reason for hiding this comment

yufansong left a comment

Choose a reason for hiding this comment

gbto commented Jul 8, 2024 •

edited by tabVersion

Loading

xxchan Jul 10, 2024 •

edited

Loading